"""
Case Type   : Logical_Replication
Case Name   : 非初始用户创建、开启和删除逻辑复制槽，验证-W参数
Create At   : 2024/1/5
Owner       : xiaqi14
Description :
    1.修改参数
    2.配置初始用户白名单
    3.创建逻辑复制槽后查询视图
    4.开启逻辑复制槽后查询视图
    5.停止逻辑复制槽后查询视图
    6.删除逻辑复制槽后查询视图
    7.恢复参数，恢复pg_hba.conf文件
Expect      :
    1.修改参数成功
    2.配置成功
    3.创建成功，视图信息正确
    4.开启成功，视图信息正确
    5.停止成功，视图信息正确
    6.删除成功，查询视图无记录
    7.恢复成功
History     :
"""
import os
import re
import unittest
from yat.test import Node
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.ComThread import ComThread


class LogicReplication(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"-----{os.path.basename(__file__)} start-----")
        self.primary_node = Node('PrimaryDbUser')
        self.sh_primary = CommonSH('PrimaryDbUser')
        self.constant = Constant()
        self.common = Common()
        self.slot_name = 'slot_logical_replication_0069'
        self.config_wal_level = 'wal_level'
        self.wal_level_default = self.common.show_param(self.config_wal_level)

    def test_logical_replication(self):
        text = "-----step1:修改参数;expect:修改参数成功-----"
        self.logger.info(text)
        if 'logical' != self.wal_level_default:
            guc_result = self.sh_primary.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_wal_level}=logical')
            self.assertTrue(guc_result, '执行失败' + text)
            restart_msg = self.sh_primary.restart_db_cluster()
            self.logger.info(restart_msg)
            self.assertTrue(restart_msg, '重启失败')
        show_msg = self.common.show_param(self.config_wal_level)
        self.assertEquals(show_msg, 'logical', '执行失败' + text)

        text = "-----step2:配置初始用户白名单;expect:配置成功-----"
        self.logger.info(text)
        param_1 = f'local replication {self.primary_node.ssh_user}  trust'
        param_2 = f'host  replication {self.primary_node.ssh_user} ' \
            f'127.0.0.1/32 trust'
        param_3 = f'host  replication {self.primary_node.ssh_user} ' \
            f'::1/128 trust'
        result_1 = self.sh_primary.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_1)
        result_2 = self.sh_primary.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_2)
        result_3 = self.sh_primary.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_3)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text)

        text = "-----step3:创建逻辑复制槽后查询视图;" \
               "expect:创建成功,视图信息正确-----"
        self.logger.info(text)
        create_slot_msg = self.sh_primary.exec_pg_recvlogical(
            '--create', 'postgres', self.slot_name,
            self.primary_node.ssh_user, '-s 2 ')
        self.logger.info(create_slot_msg)
        text = "-----查询视图;expect:视图信息正确-----"
        self.logger.info(text)
        sql_cmd = f"select * from  pg_replication_slots " \
            f"where slot_name like '{self.slot_name}';" \
            f"select * from  dbe_perf.global_replication_slots " \
            f"where slot_name like '{self.slot_name}';"
        sql_msg = self.sh_primary.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        expect_1 = "slot_name | plugin | slot_type | datoid | database | " \
                   "active | xmin | catalog_xmin | restart_lsn | dummy_standby"
        expect_2 = f"{self.slot_name} | mppdb_decoding | logical   " \
            f"|  .* | postgres | f      |      | .* | .* | f"
        expect_3 = "node_name | slot_name | plugin | slot_type | " \
                   "datoid | database | active | x_min | catalog_xmin | " \
                   "restart_lsn | dummy_standby"
        expect_4 = f"dn_6001.* |{self.slot_name} | mppdb_decoding | " \
            f"logical   |  .* | postgres | f      |      | .* | .* | f"
        search_res = re.search(expect_1, sql_msg.splitlines()[0], re.S)
        self.assertIsNotNone(search_res, '执行失败' + text)
        search_res = re.search(expect_2, sql_msg.splitlines()[2], re.S)
        self.assertIsNotNone(search_res, '执行失败' + text)
        search_res = re.search(expect_3, sql_msg.splitlines()[-4], re.S)
        self.assertIsNotNone(search_res, '执行失败' + text)
        search_res = re.search(expect_4, sql_msg.splitlines()[-2], re.S)
        self.assertIsNotNone(search_res, '执行失败' + text)

        text = "-----step4:开启逻辑复制槽后查询视图;" \
               "expect:开启成功，视图信息正确-----"
        self.logger.info(text)
        thread = ComThread(self.sh_primary.exec_pg_recvlogical,
                           args=('--start', 'postgres', self.slot_name,
                                 self.primary_node.ssh_user,
                                 f'-f - -F 1 -s 2 ',))
        thread.setDaemon(True)
        thread.start()
        thread.join(10)
        self.logger.info(thread.get_result())
        text = "-----查询视图;expect:视图信息正确-----"
        self.logger.info(text)
        sql_cmd = f"select * from  pg_replication_slots " \
            f"where slot_name like '{self.slot_name}';" \
            f"select * from  dbe_perf.global_replication_slots " \
            f"where slot_name like '{self.slot_name}';"
        sql_msg = self.sh_primary.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        expect_5 = f"{self.slot_name} | mppdb_decoding | logical   " \
            f"|  .* | postgres | t      |      | .* | .* | f"
        expect_6 = f"dn_6001.* |{self.slot_name} | mppdb_decoding | " \
            f"logical   |  .* | postgres | t      |      | .* | .* | f"
        search_res = re.search(expect_1, sql_msg.splitlines()[0], re.S)
        self.assertIsNotNone(search_res, '执行失败' + text)
        search_res = re.search(expect_5, sql_msg.splitlines()[2], re.S)
        self.assertIsNotNone(search_res, '执行失败' + text)
        search_res = re.search(expect_3, sql_msg.splitlines()[-4], re.S)
        self.assertIsNotNone(search_res, '执行失败' + text)
        search_res = re.search(expect_6, sql_msg.splitlines()[-2], re.S)
        self.assertIsNotNone(search_res, '执行失败' + text)

        text = "-----step5:停止逻辑复制槽后查询视图," \
               "expect:停止成功，视图信息正确-----"
        self.logger.info(text)
        stop_cmd = "ps -ef | grep  pg_recvlogical | grep -v grep | " \
                   "awk '{{print $2}}' | xargs kill -9"
        self.logger.info(stop_cmd)
        primary_result = self.primary_node.sh(stop_cmd).result()
        self.logger.info(primary_result)
        text = "-----查询视图;expect:视图信息正确-----"
        self.logger.info(text)
        sql_cmd = f"select * from  pg_replication_slots " \
            f"where slot_name like '{self.slot_name}';" \
            f"select * from  dbe_perf.global_replication_slots " \
            f"where slot_name like '{self.slot_name}';"
        sql_msg = self.sh_primary.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        search_res = re.search(expect_1, sql_msg.splitlines()[0], re.S)
        self.assertIsNotNone(search_res, '执行失败' + text)
        search_res = re.search(expect_2, sql_msg.splitlines()[2], re.S)
        self.assertIsNotNone(search_res, '执行失败' + text)
        search_res = re.search(expect_3, sql_msg.splitlines()[-4], re.S)
        self.assertIsNotNone(search_res, '执行失败' + text)
        search_res = re.search(expect_4, sql_msg.splitlines()[-2], re.S)
        self.assertIsNotNone(search_res, '执行失败' + text)

        text = "-----step6:删除逻辑复制槽后查询视图," \
               "expect:删除成功，查询视图无记录-----"
        self.logger.info(text)
        drop_slot_msg = self.sh_primary.exec_pg_recvlogical(
            '--drop', 'postgres', self.slot_name,
            self.primary_node.ssh_user, '-s 2 ')
        self.logger.info(drop_slot_msg)
        sql_cmd = f"select * from  pg_replication_slots " \
            f"where slot_name like '{self.slot_name}';" \
            f"select * from  dbe_perf.global_replication_slots " \
            f"where slot_name like '{self.slot_name}';"
        sql_msg = self.sh_primary.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        search_res = re.search(expect_1, sql_msg.splitlines()[0], re.S)
        self.assertIsNotNone(search_res, '执行失败' + text)
        self.assertIn("(0 rows)", sql_msg.splitlines()[2], '执行失败:' + text)
        search_res = re.search(expect_3, sql_msg.splitlines()[-3], re.S)
        self.assertIsNotNone(search_res, '执行失败' + text)
        self.assertIn("(0 rows)", sql_msg.splitlines()[-1], '执行失败:' + text)

    def tearDown(self):
        text_2 = "-----step7:恢复参数，恢复pg_hba.conf文件，" \
                 "expect:恢复成功-----"
        self.logger.info(text_2)
        param_1 = f'local replication {self.primary_node.ssh_user} '
        param_2 = f'host replication {self.primary_node.ssh_user} 127.0.0.1/32 '
        param_3 = f'host  replication {self.primary_node.ssh_user} ::1/128 '
        result_1 = self.sh_primary.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_1)
        result_2 = self.sh_primary.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_2)
        result_3 = self.sh_primary.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_3)

        wal_level_current = self.common.show_param(self.config_wal_level)
        if wal_level_current != self.wal_level_default:
            self.sh_primary.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_wal_level}={self.wal_level_default}')
            restart_msg = self.sh_primary.restart_db_cluster()
            self.logger.info(restart_msg)
        show_wal_level = self.common.show_param(self.config_wal_level)

        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text_2)
        self.assertEquals(show_wal_level, self.wal_level_default,
                          '执行失败' + text_2)
        self.logger.info(f"-----{os.path.basename(__file__)} end-----")
